package com.sad.jetpack.v1.datamodel.api.extension.client.http.sse;

import com.sad.jetpack.v1.datamodel.api.DataModelResponseImpl;
import com.sad.jetpack.v1.datamodel.api.DataSource;
import com.sad.jetpack.v1.datamodel.api.IDataModelChainOutput;
import com.sad.jetpack.v1.datamodel.api.IDataModelProductEngine;
import com.sad.jetpack.v1.datamodel.api.IDataModelRequest;
import com.sad.jetpack.v1.datamodel.api.IDataModelResponse;
import com.sad.jetpack.v1.datamodel.api.utils.CollectionTraverseUtils;

import org.apache.commons.lang3.ObjectUtils;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import kotlin.Pair;
import okhttp3.Headers;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;
import okhttp3.sse.EventSources;

public abstract class OkhttpSSEEngine extends AbsSSEEngine {

    private EventSource eventSource;

    @Override
    public void onEngineExecute(IDataModelRequest dataModelRequest, IDataModelChainOutput<String> chainOutput) throws Exception {
        Request.Builder okhttpRequestBuilder=new Request.Builder();
        okhttpRequestBuilder.url(dataModelRequest.url());
        if (dataModelRequest.method()== IDataModelRequest.Method.GET){
            okhttpRequestBuilder=okhttpRequestBuilder.get();
        }
        if (dataModelRequest.headers()!=null){
            Iterator<Map.Entry<String, String>> iterator=dataModelRequest.headers().entrySet().iterator();
            while (iterator.hasNext()){
                Map.Entry<String,String> entityEntry=iterator.next();
                okhttpRequestBuilder.addHeader(entityEntry.getKey(),entityEntry.getValue());
            }
        }
        onResetOkhttpRequest(dataModelRequest,okhttpRequestBuilder);
        OkHttpClient.Builder builder=new OkHttpClient.Builder()
                .callTimeout(dataModelRequest.timeout(), TimeUnit.MILLISECONDS)//这边需要将超时显示设置长一点，最好是一天以上，不然刚连上就断开
                .connectTimeout(dataModelRequest.timeout(), TimeUnit.MILLISECONDS)
                .followRedirects(true);
        onResetOkhttpClient(dataModelRequest,builder);
        OkHttpClient okHttpClient=builder.build();
        IDataModelResponse.Creator<String> responseCreator= DataModelResponseImpl.<String>newCreator()
                .request(dataModelRequest)
                .dataSource(DataSource.NET)
                ;
        EventSource.Factory factory = EventSources.createFactory(okHttpClient);
        EventSourceListener eventSourceListener = new EventSourceListener() {
            /**
             * {@inheritDoc}
             */
            @Override
            public void onOpen(final EventSource eventSource, final Response
                    response) {
                //log.info("建立sse连接...");
                responseCreator.code(response.code());
                Headers headers=response.headers();
                Map<String,String> h=new HashMap<>();
                Iterator<Pair<String, String>> iterator=headers.iterator();
                while (iterator.hasNext()){
                    Pair<String, String> entityEntry=iterator.next();
                    String k=entityEntry.getFirst();
                    String v=entityEntry.getSecond();
                    if (v!=null){
                        h.put(k,v);
                    }
                }
                responseCreator.headers(h);
                if (!response.isSuccessful()){
                    chainOutput.proceed(responseCreator.create());
                }
                if (getConnectionListener()!=null){
                    getConnectionListener().onSSEConnectionAlive(dataModelRequest);
                }

            }

            /**
             * {@inheritDoc}
             */
            @Override
            public void onEvent(final EventSource eventSource, final String
                    id, final String type, final String data) {
                //log.info("{}: {}", id, data);
                if (getConnectionListener()!=null){
                    getConnectionListener().onSSEDataReceived(dataModelRequest, data);
                }
                responseCreator.body(data);
                chainOutput.proceed(responseCreator.create());
            }

            /**
             * {@inheritDoc}
             */
            @Override
            public void onClosed(final EventSource eventSource) {

                //log.info("sse连接关闭...");
                if (getConnectionListener()!=null){
                    getConnectionListener().onSSEConnectionClosed(dataModelRequest);
                }
            }

            /**
             * {@inheritDoc}
             */
            @Override
            public void onFailure(final EventSource eventSource, final Throwable t, final Response response) {
                //log.error("使用事件源时出现异常... [响应：{}]...", response, t);
                if (chainOutput.exceptionListener()!=null){
                    chainOutput.exceptionListener().onDataObtainedException(dataModelRequest,t);
                }
            }
        };
        //创建事件
        this.eventSource=factory.newEventSource(okhttpRequestBuilder.build(), eventSourceListener);

    }
    public abstract void onResetOkhttpClient(IDataModelRequest request, OkHttpClient.Builder builder);
    public abstract void onResetOkhttpRequest(IDataModelRequest request, Request.Builder okhttpRequestBuilder);

    @Override
    public void close() {
        if (this.eventSource!=null){
            this.eventSource.cancel();
            //client.dispatcher().executorService().shutdown();
        }
    }
}
